package com.ruoyi.flink.chapten5;

/**
 * Copyright (c) 2020-2030 尚硅谷 All Rights Reserved
 * <p>
 * Project:  FlinkTutorial
 * <p>
 * Created by  wushengran
 */

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class Trans3FlatmapTest {
	public static void main(String[] args) throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setParallelism(1);

		DataStreamSource<Event> stream = env.fromElements(new Event("Mary", "./home", 1000L),
				new Event("Bob", "./cart", 2000L));

		
		// 1.传入一个实现了FlatMapFunction的类的对象
		SingleOutputStreamOperator<String> result1 = stream.flatMap(new MyFlatMap());
		// result1.print();

		
		// 2.使用匿名类实现FlatMapFunction
		SingleOutputStreamOperator<String> result2 = stream.flatMap(new FlatMapFunction<Event, String>() {

			@Override
			public void flatMap(Event value, Collector<String> out) throws Exception {
				if (value.user.equals("Mary")) {
					out.collect(value.user);
				} else if (value.user.equals("Bob")) {
					out.collect(value.user);
					out.collect(value.url);
				}	
			}
		});
		// result2.print("匿名调用:");
		
		
		// 3. 传入一个Lambda表达式
		stream.flatMap((Event value, Collector<String> out) -> {
			if (value.user.equals("Mary")) {
				out.collect(value.user);
			} else if (value.user.equals("Bob")) {
				out.collect(value.user);
				out.collect(value.url);
			}
		}).returns(new TypeHint<String>() {}).print();
		
		// 或 returns(Types.STRING).print();
		

		env.execute();
	}

	public static class MyFlatMap implements FlatMapFunction<Event, String> {
		@Override
		public void flatMap(Event value, Collector<String> out) throws Exception {
			if (value.user.equals("Mary")) {
				out.collect(value.user);
			} else if (value.user.equals("Bob")) {
				out.collect(value.user);
				out.collect(value.url);
			}
		}
	}
}
